/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_flush_worker.cpp
 *
 * IDENTIFICATION
 * src/gausskernel/storage/htap/htap_flush_worker.cpp
 *
 * -------------------------------------------------------------------------
 */

#include <ostream>
#include <istream>
#include <iomanip>
#include <cstring>
#include "c.h"
#include "utils/palloc.h"
#include "htap_logger.h"
#include "htap_debug.h"
#include "htap_flush_worker.h"
#include "htap_flush.h"
#include "htap_infra.h"
#include "access/imcs/imcu.h"
#include "postgres.h"
#include "storage/procarray.h"
#include "postmaster/postmaster.h"
#include "access/imcs/imcs_am.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "access/imcs/imcs_flush_insert.h"
#include "storage/remote_read.h"
#include "access/imcs/imcs_vector.h"
#include "postmaster/postmaster.h"
#include "executor/executor.h"
#include "utils/postinit.h"
#include "utils/ps_status.h"
#include "catalog/pg_database.h"
#include "catalog/pg_partition_fn.h"
#include "htap_helpers.h"
#include "htap_statistics.h"
#include "utils/atomic.h"

using namespace std;

#define bitmap_size(x) (((x) + 7) / 8) 

namespace htap {
HTAP_DECLARE_LOGGER(HTAPFlushWorker, HTAP)

static void FlushWorkerExit(SIGNAL_ARGS)
{
    int save_errno = errno;

    if (t_thrd.proc) {
        HTAP_LOG_DEBUG("Received flush worker exit signal");
        SetLatch(&t_thrd.proc->procLatch);
    }

    errno = save_errno;
}

static void FlushWorkerGotSigHup(SIGNAL_ARGS)
{
    int save_errno = errno;

    if (t_thrd.proc) {
        HTAP_LOG_DEBUG("Received flush worker hangup signal");
        SetLatch(&t_thrd.proc->procLatch);
    }

    errno = save_errno;
}

ThreadId HTAPFlushWorker::Start(int worker_num)
{
    std::lock_guard<std::mutex> guard(m_lock);
    // launch flush thread as kernel thread
    m_thread_num = worker_num;
    m_need_to_stop = false;
    HTAP_LOG_INFO("Starting flush worker thread %d", worker_num);
    PG_TRY();
    {
        m_flushWorkerThread = initialize_util_thread(IMCS_FLUSH_WORKER);
        if (m_flushWorkerThread == 0) {
            HTAP_REPORT_ERROR(HTAP_RC_SYSTEM_FAILURE, "HTAPFlushWorkerStart",
                "Failed to create kernel thread for flush");
        }
        m_isRunning = true;
    }
    PG_CATCH();
    {
        HTAP_LOG_ERROR_STACK("Failed to create flush thread: %s (%d)", Geterrmsg(), geterrcode());
        FlushErrorState();
    }
    PG_END_TRY();

    return m_flushWorkerThread;
}

RC HTAPFlushWorker::Stop()
{
    std::lock_guard<std::mutex> guard(m_lock);

    m_need_to_stop = true;
    HTAP_LOG_INFO("Stoping flush thread  %d  (%lu)", m_thread_num, m_flushWorkerThread);
    if (!m_isRunning) {
        HTAP_REPORT_ERROR(HTAP_RC_INVALID_STATE, "HTAPFlushWorker Stop",
            "Cannot stop flush worker thread: flush thread is not running");
        return HTAP_RC_INVALID_STATE;
    }
    if (t_thrd.proc) {
        SetLatch(&t_thrd.proc->procLatch);
    }
    return HTAP_RC_OK;
}

bool HTAPFlushWorker::IsRunning()
{
    std::lock_guard<std::mutex> guard(m_lock);
    return m_isRunning;
}

ThreadId HTAPFlushWorker::GetThreadId()
{
    std::lock_guard<std::mutex> guard(m_lock);
    return m_flushWorkerThread;
}

void HTAPFlushWorker::InitThread()
{
    // we are a postmaster sub-process now
    IsUnderPostmaster = true;
    t_thrd.role = IMCS_FLUSH_WORKER;

    /* reset t_thrd.proc_cxt.MyProcPid */
    t_thrd.proc_cxt.MyProcPid = gs_thread_self();

    /* record Start Time for logging */
    t_thrd.proc_cxt.MyStartTime = time(NULL);
    t_thrd.proc_cxt.MyProgName = "HTAPFlushWorker";

    /* Identify myself via ps */
    init_ps_display("flush delta tables in back-end process", "", "", "");

    SetProcessingMode(InitProcessing);

    (void)gspqsignal(SIGINT, SIG_IGN);
    (void)gspqsignal(SIGTERM, SIG_IGN);
    (void)gspqsignal(SIGQUIT, FlushWorkerExit);
    (void)gspqsignal(SIGALRM, SIG_IGN);

    (void)gspqsignal(SIGPIPE, SIG_IGN);
    (void)gspqsignal(SIGUSR1, SIG_IGN);
    (void)gspqsignal(SIGUSR2, SIG_IGN);
    (void)gspqsignal(SIGFPE, SIG_IGN);
    (void)gspqsignal(SIGHUP, FlushWorkerGotSigHup);
    (void)gspqsignal(SIGURG, SIG_IGN);

    /* Early initialization */
    BaseInit();

    /* Unblock signals (they were blocked when the postmaster forked us) */
    gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
    (void)gs_signal_unblock_sigusr2();

    t_thrd.proc_cxt.PostInit->SetDatabaseAndUser(HTAPFlush::GetInstance()->GetDbName(), InvalidOid, m_user_name);
    t_thrd.proc_cxt.PostInit->InitIMCSFlush();
}

static bool getTable(MOT::Table *table, void *data)
{
    std::list<uint64_t> *list = (std::list<uint64_t> *)data;
    list->push_back(table->GetTableExId());
    return true;
}

void HTAPFlushWorker::RunFlushWorker()
{
    bool shouldSleep = true;
    uint64_t prevFlushCSN = 0;
    uint64_t currFlushCSN = 0;
    if (m_need_to_stop) {
        m_isRunning = false;
        m_flushWorkerThread = 0;
        return;
    }
    HTAP_LOG_INFO("Flush worker %d thread %lu Running...", m_thread_num, m_flushWorkerThread);

    InitThread();

    if (!HTAPInfra::HTAPTxnEnsure(false)) {
        return;
    }

    MOT::TxnManager *txn = HTAPGetSafeTxn();
    txn->GetGcSession()->SetGcType(MOT::GcManager::GC_TYPE::GC_HTAP);
    // flush main loop
    uint64_t flushTimeoutMillis = HTAPInfra::GetInstance()->GetCfg().m_flushTimeoutMillis;

    m_flush_ctx = nullptr;
    prevFlushCSN = HTAPFlush::GetInstance()->GetLastFlushedCSN();
    while (!SignaledToStop(flushTimeoutMillis)) {
        // handle configuration change
        if (t_thrd.htap_cxt.flush_got_sighup) {
            t_thrd.htap_cxt.flush_got_sighup = false;
            ProcessConfigFile(PGC_SIGHUP);
            flushTimeoutMillis = HTAPInfra::GetInstance()->GetCfg().m_flushTimeoutMillis;
        }

        // run GC on lastFlushdCSN change only
        currFlushCSN = HTAPFlush::GetInstance()->GetLastFlushedCSN();
        if (prevFlushCSN != currFlushCSN) {
            txn->SetSnapshot();
            txn->GcSessionEnd();
            txn->SetSnapshotStatus(false);
            prevFlushCSN = currFlushCSN;
            std::list<uint64_t> table_list;
            MOT::GetTableManager()->ForEach(getTable, &table_list);
            for (uint64_t tableId : table_list) {
                MOT::Table *table = MOT::GetTableManager()->GetTableSafeByExId(tableId);
                if (table != nullptr) {
                    table->ClearThreadMemoryCache();
                    table->Unlock();
                }
            }
            table_list.clear();
            shouldSleep = false;
        }
        PerformTasks(txn);
        if (shouldSleep) {
            pg_usleep(5000L);
        } else {
            shouldSleep = true;
        }
    }
    // cleanup
    HTAP_LOG_INFO("Flush worker thread %d  (%lu) exiting", m_thread_num, m_flushWorkerThread);
    m_isRunning = false;
    m_flushWorkerThread = 0;
}

bool HTAPFlushWorker::SignaledToStop(uint64_t timeoutMillis)
{
    // wait for the given timeout for one of 2 events: latch was set, or shutdown was signaled
    bool res = false;
    int waitRes = WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT, timeoutMillis);
    if (waitRes & WL_POSTMASTER_DEATH) {
        HTAP_LOG_DEBUG("Flush detected postmaster death");
        res = true;
    } else if (m_need_to_stop) {
        HTAP_LOG_DEBUG("Flush worker %lu exit signal caught", m_flushWorkerThread);
        res = true;
    }
    return res;
}

void HTAPFlushWorker::handleDeletes(int rg_idx)
{
    CacheSlotId_t slot_id = CACHE_BLOCK_INVALID_IDX;
    IMCUDesc *cu_desc = IMCU_CACHE->GetImcuDescByHash(m_table_ctx->m_imcs_desc, rg_idx, IMCS_CTID_IDX);
    if (cu_desc == NULL) {
        HTAP_LOG_ERROR("Can't find CTID CU for rg %d in table %d. Exitting", rg_idx, m_table_ctx->m_rel_oid);
        return;
    }

    HTAPDelta::LookupContext lookupCtx;
    m_table_ctx->m_htap_delta->initLookupCtx(lookupCtx);

    if (!m_table_ctx->m_htap_delta->checkRowgroup(lookupCtx, cu_desc)) {
        gs_atomic_add_32(&m_table_ctx->m_num_filtered_rgs_2, 1);
        HTAP_LOG_DEBUG("HTAPFlushWorker: no relevant data found in htap delta for current CU for rg %d. Skipping", rg_idx);
        return;
    }

    IMCU *cu = getCU(cu_desc, IMCS_CTID_IDX, slot_id);
    if (cu == NULL) {
        HTAP_LOG_ERROR("HTAPFlushWorker: can not get ctid CU for rg %d. Skipping", rg_idx);
        return;
    }

    char foundRecordType;
    HTAPDeleteMap *rg_deleted_bitmap =
        (m_table_ctx->m_imcs_desc->deletedList) ? m_table_ctx->m_imcs_desc->deletedList[rg_idx] : NULL;

    uint max_unsigned_rows_count = cu_desc->row_count - (rg_deleted_bitmap ? rg_deleted_bitmap->bitsSigned() : 0);

    CTID min_ctid = *(CTID *)cu_desc->cu_min;
    CTID max_ctid = *(CTID *)cu_desc->cu_max;

    uint64_t startTime = GetSysClock();
    bool processingFirstUnsigned = true;
    uint32_t row_idx = (rg_deleted_bitmap ? rg_deleted_bitmap->firstUnsignedIdx() : 0);
    for (uint unsigned_rows_count = 0; unsigned_rows_count < max_unsigned_rows_count; row_idx++) {
        if (rg_deleted_bitmap && rg_deleted_bitmap->check(row_idx)) {
            HTAP_LOG_TRACE("HTAPFlushWorker: Table %d Rowgroup %d, row %d is deleted - continue to the next row",
                m_table_ctx->m_rel_oid, rg_idx, row_idx);
            continue;
        }
        unsigned_rows_count++;
        ScalarValue ctid_value = cu->GetValue<sizeof(CTID), false>(row_idx);
        CTID *ctid = (CTID *)(&ctid_value);

        HTAP_LOG_TRACE("HTAPFlushWorker: Table %d Rowgroup %d", m_table_ctx->m_rel_oid, rg_idx);

        if (m_table_ctx->m_htap_delta->lookUp(foundRecordType, lookupCtx, ctid->item)) {
            FlushStatisticsProvider::GetInstance().AddDeleteRows(1);
            m_table_ctx->m_deleted++;
            switch (foundRecordType) {
                case HTAP_DELTA_DELETE_TAG:
                    m_table_ctx->m_imcs_desc->markDeleted(rg_idx, row_idx);
                    HTAP_LOG_TRACE("HTAPFlushWorker: Setting delete bitmap for DELETE. Table %d, rg %d",
                        m_table_ctx->m_rel_oid, rg_idx);
                    break;
                case HTAP_DELTA_UPDATE_TAG:
                    m_table_ctx->m_imcs_desc->markDeleted(rg_idx, row_idx);
                    HTAP_LOG_TRACE("HTAPFlushWorker: Setting delete bitmap for UPDATE. table %d, rg %d",
                        m_table_ctx->m_rel_oid, rg_idx);
                    break;
                case HTAP_DELTA_INSERT_TAG:
                    HTAP_LOG_TRACE("HTAPFlushWorker: %s is not expected to have INSERTED status during flush. Setting "
                                   "delete bitmap, as this cid will be shortly added ",
                        m_table_ctx->m_table->GetTableName().c_str());
                    m_table_ctx->m_imcs_desc->markDeleted(rg_idx, row_idx);
                    break;
                default:
                    HTAP_LOG_ERROR("HTAPFlushWorker: delta hash: unknown operation %d", foundRecordType);
            }
        } else {
            // Note - for updating firstUnsigned we use the fact that all cases (except for default, which we ignore),
            // are setting delete bit
            if (processingFirstUnsigned) {
                rg_deleted_bitmap = (m_table_ctx->m_imcs_desc->deletedList) ?
                    m_table_ctx->m_imcs_desc->deletedList[rg_idx] :
                    NULL;
                if (rg_deleted_bitmap)
                    rg_deleted_bitmap->updateFirstUnsignedIdx(row_idx);
                processingFirstUnsigned = false;
                min_ctid = max_ctid = *ctid;
            } else {
                if (ItemPointerCompare(&ctid->item, &((CTID *)&min_ctid)->item) < 0) {
                    min_ctid = *ctid;
                } else if (ItemPointerCompare(&ctid->item, &((CTID *)&max_ctid)->item) > 0) {
                    max_ctid = *ctid;
                }
            }
        }
    }
    // Update min max
    if (ItemPointerCompare(&((CTID *)cu_desc->cu_min)->item, &min_ctid.item) != 0) {
        if (IMCU_CACHE->AcquireImcsRgLock(m_table_ctx->m_rel_oid, LW_EXCLUSIVE, rg_idx)) {
            *(CTID *)cu_desc->cu_min = min_ctid;
            IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, rg_idx);
        } else {
            HTAP_LOG_WARN("HTAPFlushWorker: execDeletes update min for table %d rg %d - couldn't acquire rg_lock, "
                          "skip min update",
                m_table_ctx->m_rel_oid, rg_idx);
        }
    }
    if (ItemPointerCompare(&((CTID *)cu_desc->cu_max)->item, &max_ctid.item) != 0) {
        if (IMCU_CACHE->AcquireImcsRgLock(m_table_ctx->m_rel_oid, LW_EXCLUSIVE, rg_idx)) {
            *(CTID *)cu_desc->cu_max = max_ctid;
            IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, rg_idx);
        } else {
            HTAP_LOG_WARN("HTAPFlushWorker: execDeletes update max for table %d rg %d - couldn't acquire rg_lock, "
                          "skip max update",
                m_table_ctx->m_rel_oid, rg_idx);
        }
    }
    uint64_t endTime = GetSysClock();
    uint64_t timeMicros = MOT::CpuCyclesLevelTime::CyclesToMicroseconds(endTime - startTime);
    gs_atomic_add_64(&m_table_ctx->m_delete_time, timeMicros);
    if (IsValidCacheSlotID(slot_id))
        IMCU_CACHE->UnPinDataBlock(slot_id);
}

void HTAPFlushWorker::PerformTasks(MOT::TxnManager *txn)
{
    FlushTaskQueue *working_queue = HTAPInfra::GetInstance()->GetFlush()->getWorkingQueue();
    FlushTask *task = working_queue->Get();
    if (task != nullptr) {
        m_flush_ctx = HTAPInfra::GetInstance()->CreateMemoryContext(
            HTAPInfra::GetInstance()->GetThreadMemoryContext(), "flushCtx");
        MemoryContext saved_cycle_context = MemoryContextSwitchTo(m_flush_ctx);

        StartTransactionCommand();
        Snapshot snap = RegisterSnapshot(GetTransactionSnapshot());
        PushActiveSnapshot(snap);
        m_table_ctx = task->table_ctx;
        PG_TRY();
        {
            switch (task->task_type) {
                case DELTA_DELETE:
                    HTAP_LOG_DEBUG("Received DELTA_DELETE task %p: table %d rg %d", task, m_table_ctx->m_rel_oid,
                        task->rg_idx);
                    handleDeletes(task->rg_idx);
                    break;

                case DELTA_INSERT: {
                    HTAPDelta::ScanContext scanCtx(task->opaque);
                    HTAP_LOG_DEBUG("Received DELTA_INSERT task %p: table %d", task, m_table_ctx->m_rel_oid);
                    m_imcs_insert = New(m_flush_ctx)IMCSFlushInsert(m_table_ctx->m_rel,
                        m_table_ctx->m_result_rel_info, false, NULL, NULL, m_table_ctx->m_tuple_desc);
                    addToRowgroups(scanCtx, true); // true for multithread
                    DELETE_EX(m_imcs_insert);
                    break;
                }

                case DELTA_COMPLETE:
                    HTAPDelta::ScanContext scanCtx(task->opaque);
                    completeInserts(scanCtx, task->rg_idx, task->free_rows);
                    break;
            }
        }
        PG_CATCH();
        {
            HTAP_LOG_ERROR(
                "HTAPFlushWorker: internal unknown issue while working on table %d, task_type %d rg %d. Details: %s",
                m_table_ctx->m_rel_oid, task->task_type, task->rg_idx, t_thrd.log_cxt.error_context_stack);
            FlushErrorState();
        }
        PG_END_TRY();

        HTAP_LOG_DEBUG("Finished task %p: reducing table %d ref count %d by 1", task, task->table_ctx->m_rel_oid,
            task->table_ctx->getRefCount());
        task->table_ctx->updateRefCount(-1);
        working_queue->SetFinished(task);
        PopActiveSnapshot();
        UnregisterSnapshot(snap);
        CommitTransactionCommand();

        MemoryContextSwitchTo(saved_cycle_context);
        HTAPInfra::GetInstance()->DestroyMemoryContext(m_flush_ctx);
    }
}

RC HTAPFlushWorker::getOldCUs(IMCU **old_cus, TupleDesc tuple_desc_for_old_cus, IMCUDesc **imcu_descs_array,
    CacheSlotId_t *slot_array)
{
    int col_num = tuple_desc_for_old_cus->natts;
    for (int col_idx = 0; col_idx < col_num; ++col_idx) {
        CacheSlotId_t slot_id = CACHE_BLOCK_INVALID_IDX;
        if (imcu_descs_array[col_idx]->IsNullCU())
            old_cus[col_idx] = nullptr;
        else {
            old_cus[col_idx] = getCU(imcu_descs_array[col_idx], col_idx, slot_id);
        }
        if ((old_cus[col_idx] == NULL) && (!imcu_descs_array[col_idx]->IsNullCU())) {
            HTAP_LOG_ERROR("HTAPFlushWorker:getOldCUs CU not found for col_idx %d table %d", col_idx,
                m_table_ctx->m_imcs_desc->imcs_oid);
            return HTAP_RC_ERROR;
        }
        slot_array[col_idx] = slot_id;
    }

    return HTAP_RC_OK;
}

IMCUDesc **HTAPFlushWorker::initImcuDescsArray(int rg_idx)
{
    int col_num = m_table_ctx->m_tuple_desc->natts;
    IMCUDesc **imcu_descs_array = (IMCUDesc **)palloc0(sizeof(IMCUDesc *) * (col_num));
    for (int col_idx = 0; col_idx < col_num; ++col_idx) {
        int attnum = (col_idx == 0) ? 0 : m_table_ctx->m_tuple_desc->attrs[col_idx - 1].attnum;
        imcu_descs_array[col_idx] = IMCU_CACHE->GetImcuDescByHash(m_table_ctx->m_imcs_desc, rg_idx, attnum);
        if (imcu_descs_array[col_idx] == NULL) {
            HTAP_LOG_ERROR("HTAPFlushWorker: Can't find CU Desc for rg %d col %d in table %d. Exitting", rg_idx,
                col_idx, m_table_ctx->m_imcs_desc->imcs_oid);
            return NULL;
        }
        // identify the last CU and verify it is partial
        if ((imcu_descs_array[col_idx]->m_cu == NULL) && (!imcu_descs_array[col_idx]->IsNullCU())) {
            HTAP_LOG_ERROR("HTAPFlushWorker: CU is null for rg %d col %d in table %d. Exitting", rg_idx, col_idx,
                m_table_ctx->m_imcs_desc->imcs_oid);
            return NULL;
        }
    }
    return imcu_descs_array;
}

void HTAPFlushWorker::addToRowgroups(HTAPDelta::ScanContext &scan_ctx, bool multithread, int free_rows_cur_rg,
    IMCU **old_cus, IMCUDesc **old_cus_descs)
{
    HTAP_LOG_DEBUG("HTAPFlushWorker:addInserts: Add inserts to a new RG rg %d, table %d",
        m_table_ctx->m_imcs_desc->rowgroup_num, m_table_ctx->m_rel_oid);
    int col_num = m_table_ctx->m_tuple_desc->natts;
    Datum *values = (Datum *)palloc0(col_num * sizeof(Datum));
    bool *nulls = (bool *)palloc0(col_num * sizeof(bool));
    size_t *lengths = (size_t *)palloc0(col_num * sizeof(size_t));

    IMCSBulkloadRows *batch_rows_ptr = New(m_flush_ctx)
        IMCSBulkloadRows(m_table_ctx->m_tuple_desc, min(free_rows_cur_rg, IMCS_MAX_ROWS_SIZE), true);
    uint64_t startTime = GetSysClock();

    m_table_ctx->m_htap_delta->scanForFlush(batch_rows_ptr, values, nulls, lengths, m_table_ctx->m_imcs_to_values,
        m_table_ctx->m_tuple_desc, scan_ctx, min(free_rows_cur_rg, IMCS_MAX_ROWS_SIZE));

    if (batch_rows_ptr->m_rows_curnum > 0) {
        FlushStatisticsProvider::GetInstance().AddToNewRows(batch_rows_ptr->m_rows_curnum);
        m_table_ctx->m_num_rows += batch_rows_ptr->m_rows_curnum;
        m_imcs_insert->BatchInsertCommon(batch_rows_ptr, old_cus, old_cus_descs, multithread, m_table_ctx->m_imcs_desc);
        HTAP_LOG_DEBUG(
            "HTAPFlushWorker:collectInsetrsBatch Finishing - Added %d rows to table %d multithread %t rg %d",
            batch_rows_ptr->m_rows_curnum, m_table_ctx->m_rel_oid, multithread,
            multithread ? m_table_ctx->m_imcs_desc->rowgroup_num :
                          m_imcs_insert->m_imi_cur_cuid - (old_cus_descs ? 1 : 0));
    }

    uint64_t endTime = GetSysClock();
    uint64_t timeMicros = MOT::CpuCyclesLevelTime::CyclesToMicroseconds(endTime - startTime);
    gs_atomic_add_64(&m_table_ctx->m_insert_time, timeMicros);

    pfree(values);
    pfree(nulls);
    pfree(lengths);
    DELETE_EX(batch_rows_ptr);
}

void HTAPFlushWorker::setMinMax(IMCUDesc *cu_desc, bool *first, Oid atttypid, int attlen, char *val_addr)
{
    ImcsFuncSetMinMax min_max_func = ImcsGetMinMaxFunc(atttypid);
    Datum val;
    if (!min_max_func)
        return;

    if (attlen > 0 && attlen <= 8) {
        val = *(Datum *)(val_addr);
    } else {
        val = PointerGetDatum(val_addr);
    }
    min_max_func(val, cu_desc, first);
}

void HTAPFlushWorker::updateCompactCu(IMCU *cu, IMCUDesc *cu_desc, HTAPDeleteMap *delMap, Oid type_oid, int attlen,
    int atttypid)
{
    // First compact the CU's nulls to get the new nulls size (int bytes)
    char old_nulls[cu->m_bpNullRawSize];
    int nulls_size = 0;
    int pre_compact_null_size = cu->m_bpNullRawSize;
    if (cu->m_bpNullRawSize) {
        // Save the current null bit map before compation for later usage
        memcpy(old_nulls, cu->m_nulls, cu->m_bpNullRawSize);
        // Zero the null bit map before compaction of nulls
        memset(cu->m_nulls, 0, cu->m_bpNullRawSize);
        // Handle the null compaction and get the new null bit map size
        nulls_size = handleCompactNulls(cu, cu_desc, delMap, old_nulls);
    }

    char *dest_src_buf_pos = &cu->m_srcBuf[nulls_size];
    char *src_src_buf_pos = cu->m_srcData;
    Size data_size = cu->m_eachValSize;
    int deleted_cnt = 0;
    uint32 removed_data_size = 0;

    // Zero the min max since no longer relevant. and recalculate during data compaction
    bool first = true;
    memset_s(cu_desc->cu_min, MIN_MAX_LEN, 0, MIN_MAX_LEN);
    memset_s(cu_desc->cu_max, MIN_MAX_LEN, 0, MIN_MAX_LEN);
    for (int i = 0; i < cu_desc->row_count; i++) {
        bool isDeleted = false;
        // Count the deleted rows
        if (delMap->check(i)) {
            isDeleted = true;
            deleted_cnt++;
        }
        // Skip this row if it is null. nulls already handled by 'handleCompactNulls'
        if (pre_compact_null_size && old_nulls[i / 8] & (1 << (i % 8)))
            continue;
        // Get the data size in case of varlen
        if (attlen == -1) {
            data_size = VARSIZE_ANY(src_src_buf_pos);
        }
        if (attlen == -2) {
            data_size = strlen(src_src_buf_pos) + 1;
        }
        // Skip this data if deleted and move to the next data to copy
        if (isDeleted) {
            src_src_buf_pos = &src_src_buf_pos[data_size];
            removed_data_size += data_size;
            continue;
        }
        // Move the data to the new position in the buffer
        memmove(dest_src_buf_pos, src_src_buf_pos, data_size);

        // Set the min max for the current data
        setMinMax(cu_desc, &first, atttypid, attlen, dest_src_buf_pos);

        // Set the new position for src and destination in the buffer
        src_src_buf_pos = &src_src_buf_pos[data_size];
        dest_src_buf_pos = &dest_src_buf_pos[data_size];
    }

    // Set the new data size and row count in CU and descriptor
    cu_desc->row_count -= deleted_cnt;
    cu->m_srcDataSize -= removed_data_size;
    cu->m_srcData = &cu->m_srcBuf[nulls_size];

    // Update the CU mode in the descritor
    ImcsFuncSetMinMax min_max_func = ImcsGetMinMaxFunc(atttypid);
    cu_desc->cu_mode = 0;
    (void)IMCStore::SetCudescModeForMinMaxVal(false, min_max_func != nullptr, nulls_size == 0 ? false : true, -1,
        attlen, cu_desc);
    if (cu_desc->row_count == 0) {
        cu_desc->cu_mode &= ~CU_SAME_VAL;
        cu_desc->cu_mode &= ~CU_HAS_NULL;
        cu->m_infoMode &= ~CU_HasNULL;
        if (cu_desc->cu_mode != CU_NO_MINMAX_CU)
            cu_desc->SetNormalCU();
    }

    // Recalculation of offsert array
    if (cu->m_offset) {
        CStoreMemAlloc::Pfree(cu->m_offset, !cu->m_inCUCache);
    }
    cu->m_offsetSize = 0;
    cu->m_offset = NULL;
    (cu->HasNullValue()) ? cu->FormValuesOffset<true>(cu_desc->row_count) :
                           cu->FormValuesOffset<false>(cu_desc->row_count);
}

int HTAPFlushWorker::handleCompactNulls(IMCU *cu, IMCUDesc *cu_desc, HTAPDeleteMap *delMap, char *old_nulls)
{
    if (!cu->HasNullValue())
        return 0;

    uint32 pos = 0;
    uint32 old_shift, old_mask, new_shift, new_mask, val;
    // Going over all the nulss and update according to deleted
    for (int i = 0; i < cu_desc->row_count; i++) {
        // Check if this bit is not set and copy it to its new location;
        if (!delMap->check(i)) {
            old_shift = i % 8;
            old_mask = 1 << old_shift;
            new_shift = pos % 8;
            new_mask = 1 << new_shift;
            val = (old_nulls[i / 8] & old_mask) != 0;
            if (val)
                cu->m_nulls[pos / 8] |= new_mask;
            else
                cu->m_nulls[pos / 8] &= ~new_mask;
            pos++;
        }
    }

    int nulls_size = bitmap_size(pos);
    if (pos == 0) {
        // All nulls deleted - Unset 'has nulls' indications
        cu->m_infoMode &= ~CU_HasNULL;
        cu_desc->SetCUHasNull(false);
        cu->m_nulls = 0;
        cu->m_bpNullRawSize = 0;
    } else {
        cu->m_bpNullRawSize = nulls_size;
    }
    return nulls_size;
}

void HTAPFlushWorker::updateCompactIMCS(IMCSDesc *imcs_desc, int rg)
{
    m_table_ctx->m_imcs_desc->deletedList[rg]->Destroy();
    m_table_ctx->m_imcs_desc->deletedList[rg] = nullptr;
}

void HTAPFlushWorker::handleCompactForNullCu(HTAPDeleteMap *delMap, IMCUDesc *cu_desc)
{
    int deleted_cnt = 0;
    for (int i = 0; i < cu_desc->row_count; i++) {
        // Count the number of deleted rows
        if (delMap->check(i)) {
            deleted_cnt++;
        }
    }
    // Update the number of rows in the descritor.
    cu_desc->row_count -= deleted_cnt;
}

void HTAPFlushWorker::compact(int rg)
{
    if (!IMCU_CACHE->AcquireImcsRgLock(m_table_ctx->m_rel_oid, LW_EXCLUSIVE, rg))
        return;
    IMCUDesc **cus_descs = initImcuDescsArray(rg);
    if (!cus_descs) {
        HTAP_LOG_ERROR("HTAPFlushWorker:compact: cus_descs is NULL failed to compact for RG %d", rg);
        IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, rg);
        return;
    }
    IMCSDesc *imcs_desc = m_table_ctx->m_imcs_desc;
    HTAPDeleteMap *delMap =
        m_table_ctx->m_imcs_desc->deletedList ? m_table_ctx->m_imcs_desc->deletedList[rg] : nullptr;
    if (delMap) {
        HTAP_LOG_DEBUG("HTAPFlushWorker:compact: Found %d deleted rows. Starting compaction for RG %d",
            delMap->bitsSigned(), rg);
        int col_num = m_table_ctx->m_tuple_desc->natts;
        IMCU *old_cus[col_num];
        CacheSlotId_t slot_array[col_num];
        if (getOldCUs(old_cus, m_table_ctx->m_tuple_desc, cus_descs, slot_array) == HTAP_RC_OK) {
            for (int col = 0; col < col_num; col++) {
                Assert(cus_descs[col]);
                if (cus_descs[col]->IsNullCU()) {
                    handleCompactForNullCu(delMap, cus_descs[col]);
                    if (IsValidCacheSlotID(slot_array[col]))
                        IMCU_CACHE->UnPinDataBlock(slot_array[col]);
                    continue;
                }
                Assert(old_cus[col]);
                int rcol = ((col == 0) ? (col_num - 1) : (col - 1));
                updateCompactCu(old_cus[col], cus_descs[col], delMap, m_table_ctx->m_tuple_desc->attrs[rcol].atttypid,
                    m_table_ctx->m_tuple_desc->attrs[rcol].attlen, m_table_ctx->m_tuple_desc->attrs[rcol].atttypid);
                if (IsValidCacheSlotID(slot_array[col]))
                    IMCU_CACHE->UnPinDataBlock(slot_array[col]);
                HTAP_LOG_INFO("IN_CACHE_2 m_inCUCache %d", old_cus[col]->m_inCUCache);
            }
            updateCompactIMCS(imcs_desc, rg);
        } else {
            HTAP_LOG_ERROR("HTAPFlushWorker:compact: Failed to get old CUs for RG %d", rg);
        }
    }
    IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, rg);
}

void HTAPFlushWorker::completeInserts(HTAPDelta::ScanContext &scan_ctx, int complete_rg_idx, int free_rows)
{
    IMCUDesc **old_cus_descs = nullptr;
    if (complete_rg_idx >= 0) {
        // Get the IMCUDescs of the last RG
        compact(complete_rg_idx);
        if (IMCU_CACHE->AcquireImcsRgLock(m_table_ctx->m_rel_oid, LW_EXCLUSIVE, complete_rg_idx)) {
            old_cus_descs = initImcuDescsArray(complete_rg_idx);
            if (old_cus_descs) {
                Assert(free_rows == (IMCS_MAX_ROWS_SIZE - old_cus_descs[0]->row_count));
                int col_num = m_table_ctx->m_tuple_desc->natts;
                // Get the CUs of the given RG
                IMCU *old_cus[col_num];
                CacheSlotId_t slot_array[col_num];
                if (getOldCUs(old_cus, m_table_ctx->m_tuple_desc, old_cus_descs, slot_array) == HTAP_RC_OK) {
                    m_imcs_insert = New(m_flush_ctx)IMCSFlushInsert(m_table_ctx->m_rel,
                        m_table_ctx->m_result_rel_info, false, NULL, NULL, m_table_ctx->m_tuple_desc);
                    m_imcs_insert->m_imi_cur_cuid = complete_rg_idx;
                    int rows_before = old_cus_descs[0]->row_count;
                    addToRowgroups(scan_ctx, false, free_rows, old_cus, old_cus_descs);
                    int rows_after = old_cus_descs[0]->row_count;
                    HTAP_LOG_DEBUG("HTAPFlushWorker:addInserts: Complete old RG %d for table %d added %d rows",
                        complete_rg_idx, m_table_ctx->m_rel_oid, rows_after - rows_before);
                    for (int i = 0; i < col_num; i++) {
                        if (IsValidCacheSlotID(slot_array[i]))
                            IMCU_CACHE->UnPinDataBlock(slot_array[i]);
                    }
                    IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, complete_rg_idx);
                    DELETE_EX(m_imcs_insert);
                    return;
                }
                HTAP_LOG_ERROR("HTAPFlushWorker:addInserts: Failed to get old CUs for RG %d, table %d",
                    complete_rg_idx, m_table_ctx->m_rel_oid);
            }
            IMCU_CACHE->ReleaseImcsRgLock(m_table_ctx->m_rel_oid, complete_rg_idx);
        }
    }

    m_imcs_insert = New(m_flush_ctx)IMCSFlushInsert(m_table_ctx->m_rel, m_table_ctx->m_result_rel_info, false,
        NULL, NULL, m_table_ctx->m_tuple_desc);
    addToRowgroups(scan_ctx, true, free_rows);
    DELETE_EX(m_imcs_insert);
}

IMCU *HTAPFlushWorker::getCU(IMCUDesc *cu_desc_ptr, int col_idx, CacheSlotId_t &slot_id)
{
    int reverse_col_idx = (col_idx == IMCS_CTID_IDX) ? m_table_ctx->m_tuple_desc->natts - 1 : col_idx - 1;
    int attnum = (col_idx == IMCS_CTID_IDX) ? 0 : m_table_ctx->m_tuple_desc->attrs[col_idx - 1].attnum;
    DataSlotTag data_slot_tag = IMCU_CACHE->InitCUSlotTag((RelFileNodeOld *)&m_table_ctx->m_rel->rd_node, attnum,
                                                          cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer);

    bool has_found = false;
    Form_pg_attribute cur_attr = &m_table_ctx->m_tuple_desc->attrs[reverse_col_idx];
    CUUncompressedRetCode ret_code = CU_OK;
    CFileNode cfile_node(m_table_ctx->m_rel->rd_node, attnum, MAIN_FORKNUM);
    IMCUStorage *cu_storage = New(m_flush_ctx)IMCUStorage(cfile_node);

RETRY_LOAD_CU:

    /* Look for the IMCU in the cache first, this is quick and
     * should succeed most of the time.
     */
    slot_id = IMCU_CACHE->FindDataBlock(&data_slot_tag, true);
    IMCU *cu_ptr = NULL;

    // If the IMCU is not in the cache, reserve it.
    if (IsValidCacheSlotID(slot_id)) {
        has_found = true;
        // IMCU* prev_cu_ptr = (IMCU*)(cu_desc_ptr->m_cu);
    } else {
        has_found = false;
        slot_id = IMCU_CACHE->ReserveDataBlock(&data_slot_tag, cu_desc_ptr->cu_size, has_found);
        /* Use the cached IMCU */
    }
    cu_ptr = IMCU_CACHE->GetCUBuf(slot_id);
    cu_ptr->m_inCUCache = true;
    cu_ptr->SetAttInfo(cur_attr->attlen, cur_attr->atttypmod, cur_attr->atttypid);

    /* If the IMCU was already in the cache, return it. */
    if (has_found) {
        /* Wait for a read to complete, if still in progress */
        if (IMCU_CACHE->DataBlockWaitIO(slot_id)) {
            IMCU_CACHE->UnPinDataBlock(slot_id);
            HTAP_LOG_INFO("CU wait IO find an error, need to reload! table(%s), column(%s), relfilenode(%u/%u/%u), "
                          "cuid(%u)",
                          RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                          m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                          m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
            goto RETRY_LOAD_CU;
        }

        if (!cu_ptr->m_cache_compressed) {
            checkConsistenceOfCuData(cu_desc_ptr, cu_ptr, (AttrNumber)(cur_attr->attnum));
            return cu_ptr;
        }
        if (cu_ptr->m_cache_compressed) {
            ret_code = IMCU_CACHE->StartUncompressCU(cu_desc_ptr, slot_id, 0, false, ALIGNOF_CUSIZE);
            if (ret_code == CU_RELOADING) {
                IMCU_CACHE->UnPinDataBlock(slot_id);
                HTAP_LOG_INFO("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                              "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                              RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                              m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                              m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
                goto RETRY_LOAD_CU;
            } else if (ret_code == CU_ERR_ADIO) {
                HTAP_LOG_ERROR("Load CU failed in adio! table(%s), column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                               RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                               m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                               m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
            } else if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
                if (RelationNeedsWAL(m_table_ctx->m_rel) && CanRemoteRead()) {
                    /* clear ImcuCacheBlockInProgressIO and ImcuCacheBlockInProgressUncompress but not free cu buffer */
                    IMCU_CACHE->TerminateCU(false);

                    HTAP_LOG_WARN("Data corrupted - invalid CU in cu_id %u of relation %s file %s offset %lu, ret code "
                                  "%d, try to remote read",
                                  cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel), relcolpath(cu_storage),
                                  cu_desc_ptr->cu_pointer, ret_code);

                    /* remote load cu */
                    ret_code = getCuDataFromRemote(cu_desc_ptr, cu_ptr, cur_attr->attnum, cur_attr->attlen, slot_id,
                                                   cu_storage);
                    if (ret_code == CU_RELOADING) {
                        /* other thread in remote read */
                        IMCU_CACHE->UnPinDataBlock(slot_id);
                        HTAP_LOG_INFO("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                                      "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                                      RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                                      m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                                      m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
                        goto RETRY_LOAD_CU;
                    }
                } else {
                    /* unlogged table can not remote read */
                    IMCU_CACHE->TerminateCU(true);
                    HTAP_LOG_ERROR("Data corrupted - invalid CU in cu_id %u of relation %s file %s offset %lu, "
                                   "ret_code %d try to rmote read",
                                   cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel), relcolpath(cu_storage),
                                   cu_desc_ptr->cu_pointer, ret_code);
                }
            } else {
                Assert(ret_code == CU_OK);
            }
        }

        checkConsistenceOfCuData(cu_desc_ptr, cu_ptr, (AttrNumber)(cur_attr->attnum));
        DELETE_EX(cu_storage);
        return cu_ptr;
    }

    cu_storage->LoadIMCU(cu_ptr, cu_desc_ptr->cu_id, cu_desc_ptr->cu_size,
                         g_instance.attr.attr_storage.enable_adio_function, true);

    IMCU_CACHE->IncreaseIMCUDesc(m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id, cur_attr->attnum);

    /* Mark the IMCU as no longer io busy, and wake any waiters */
    IMCU_CACHE->DataBlockCompleteIO(slot_id);

    ret_code = IMCU_CACHE->StartUncompressCU(cu_desc_ptr, slot_id, 0, false, ALIGNOF_CUSIZE);
    if (ret_code == CU_RELOADING) {
        IMCU_CACHE->UnPinDataBlock(slot_id);
        HTAP_LOG_INFO("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), column(%s), "
                      "relfilenode(%u/%u/%u), cuid(%u)",
                      RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                      m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                      m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
        goto RETRY_LOAD_CU;
    } else if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
        /* Sync load IMCU contains incorrect checksum */

        if (RelationNeedsWAL(m_table_ctx->m_rel) && CanRemoteRead()) {
            /* clear ImcuCacheBlockInProgressIO and ImcuCacheBlockInProgressUncompress but not free cu buffer */
            IMCU_CACHE->TerminateCU(false);
            HTAP_LOG_WARN("invalid CU in cu_id %u of relation %s file %s offset %lu, ret_code %d, try to remote read",
                          cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel), relcolpath(cu_storage),
                          cu_desc_ptr->cu_pointer, ret_code);

            /* remote load cu */
            ret_code =
                getCuDataFromRemote(cu_desc_ptr, cu_ptr, cur_attr->attnum, cur_attr->attlen, slot_id, cu_storage);
            if (ret_code == CU_RELOADING) {
                /* other thread in remote read */
                IMCU_CACHE->UnPinDataBlock(slot_id);
                HTAP_LOG_DEBUG("The CU is being reloaded by remote read thread. Retry to load CU! table(%s), "
                               "column(%s), relfilenode(%u/%u/%u), cuid(%u)",
                               RelationGetRelationName(m_table_ctx->m_rel), NameStr(cur_attr->attname),
                               m_table_ctx->m_rel->rd_node.spcNode, m_table_ctx->m_rel->rd_node.dbNode,
                               m_table_ctx->m_rel->rd_node.relNode, cu_desc_ptr->cu_id);
                goto RETRY_LOAD_CU;
            }
        } else {
            /* unlogged table can not remote read */
            IMCU_CACHE->TerminateCU(true);
            HTAP_LOG_ERROR("Data corrupted - invalid CU in cu_id %u of relation %s file %s offset %lu, ret_code %d",
                           cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel), relcolpath(cu_storage),
                           cu_desc_ptr->cu_pointer, ret_code);
        }
    }

    Assert(ret_code == CU_OK);

    checkConsistenceOfCuData(cu_desc_ptr, cu_ptr, (AttrNumber)(cur_attr->attnum));
    DELETE_EX(cu_storage);
    return cu_ptr;
}

CUUncompressedRetCode HTAPFlushWorker::getCuDataFromRemote(IMCUDesc *cu_desc_ptr, IMCU *cu_ptr, int col_idx,
                                                           int val_size, const int &slot_id, IMCUStorage *cu_storage)
{
    FormData_pg_attribute *attrs = m_table_ctx->m_tuple_desc->attrs;
    CUUncompressedRetCode ret_code = CU_OK;

    /* reuse memory and check if have some other session is updating it concurrently. */
    if (IMCU_CACHE->ReserveCstoreDataBlockWithSlotId(slot_id)) {
        cu_ptr = IMCU_CACHE->GetCUBuf(slot_id);
        cu_ptr->m_inCUCache = true;
        cu_ptr->SetAttInfo(val_size, attrs[col_idx].atttypmod, attrs[col_idx].atttypid);

        /*
         * remote load need IMCU compressed. (cu_ptr->m_compressed_load_buf != NULL)
         * if IMCU uncompressed, means  other thread remote read cu already and uncompress it.
         */
        IMCU_CACHE->AcquireCompressLock(slot_id);

        if (cu_ptr->m_cache_compressed) {
            cu_storage->RemoteLoadCU(cu_ptr, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size,
                                     g_instance.attr.attr_storage.enable_adio_function, true);

            if (cu_ptr->IsVerified(cu_desc_ptr->magic))
                cu_storage->OverwriteCU(cu_ptr->m_compressedBuf, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size, false);
        }

        IMCU_CACHE->ReleaseCompressLock(slot_id);

        IMCU_CACHE->DataBlockCompleteIO(slot_id);
    } else {
        if (IMCU_CACHE->DataBlockWaitIO(slot_id)) {
            HTAP_LOG_ERROR("There is an IO error when remote read CU in cu_id %u of relation %s file %s offset %lu. "
                           "slot_id %d, column \"%s\" ",
                           cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel),
                           col_idx, // relcolpath(m_cu_storage[col_idx]),
                           cu_desc_ptr->cu_pointer, slot_id, NameStr(attrs[col_idx].attname));
        }
    }

    ret_code = IMCU_CACHE->StartUncompressCU(cu_desc_ptr, slot_id, 0, false, ALIGNOF_CUSIZE);
    if (ret_code == CU_ERR_CRC || ret_code == CU_ERR_MAGIC) {
        /* remote load crc error */
        IMCU_CACHE->TerminateCU(true);
        HTAP_LOG_ERROR("Data corrupted - invalid CU in cu_id %u of relation %s file %s offset %lu, error %d",
                       cu_desc_ptr->cu_id, RelationGetRelationName(m_table_ctx->m_rel),
                       col_idx, // relcolpath(m_cu_storage[col_idx]),
                       cu_desc_ptr->cu_pointer, ret_code);
    }

    return ret_code;
}

void HTAPFlushWorker::checkConsistenceOfCuData(IMCUDesc *cu_desc_ptr, IMCU *cu, AttrNumber col) const
{
    /*
     * This memory barrier prevents unordered read, which may cause using NOT-uncompress-completed IMCU.
     * We must add memory barrier before returning cu_ptr in every branch of function get_cu_data.
     */
#ifdef __aarch64__
    pg_memory_barrier();
#endif

    /* check the src data ptr. */
    if (cu->m_srcData == NULL) {
        HTAP_LOG_ERROR("HTAPFlushWorker: bad CU: src_data is NULL: relation info: name \"%s\", namespace id %u, id "
                       "%u, relfilenode %u/%u/%u",
                       RelationGetRelationName(m_table_ctx->m_rel), RelationGetNamespace(m_table_ctx->m_rel),
                       RelationGetRelid(m_table_ctx->m_rel), m_table_ctx->m_rel->rd_node.spcNode,
                       m_table_ctx->m_rel->rd_node.dbNode, m_table_ctx->m_rel->rd_node.relNode);
        HTAP_LOG_ERROR("HTAPFlushWorker: CU info: table column %d, id %u, offset %lu, size %d, row count %d", col,
                       cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size, cu_desc_ptr->row_count);
        Assert(0);
    }

    /* check the offset ptr. */
    if ((cu->m_eachValSize < 0 && cu->m_offset == NULL) || (cu->HasNullValue() && cu->m_offset == NULL)) {
        HTAP_LOG_ERROR("HTAPFlushWorker: bad CU: m_offset is NULL: relation info: name \"%s\", namespace id %u, id "
                       "%u, relfilenode %u/%u/%u",
                       RelationGetRelationName(m_table_ctx->m_rel), RelationGetNamespace(m_table_ctx->m_rel),
                       RelationGetRelid(m_table_ctx->m_rel), m_table_ctx->m_rel->rd_node.spcNode,
                       m_table_ctx->m_rel->rd_node.dbNode, m_table_ctx->m_rel->rd_node.relNode);
        HTAP_LOG_ERROR("HTAPFlushWorker: CU info: table column %d, id %u, offset %lu, size %d, row count %d", col,
                       cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size, cu_desc_ptr->row_count);
        Assert(0);
    }

    /* check the magic number */
    if (cu->m_magic != cu_desc_ptr->magic) {
        HTAP_LOG_ERROR("HTAPFlushWorker: bad magic in CU: relation info: name \"%s\", namespace id %u, id %u, "
                       "relfilenode %u/%u/%u",
                       RelationGetRelationName(m_table_ctx->m_rel), RelationGetNamespace(m_table_ctx->m_rel),
                       RelationGetRelid(m_table_ctx->m_rel), m_table_ctx->m_rel->rd_node.spcNode,
                       m_table_ctx->m_rel->rd_node.dbNode, m_table_ctx->m_rel->rd_node.relNode);
        HTAP_LOG_ERROR("HTAPFlushWorker: CU info: table column %d, id %u, offset %lu, size %d, row count %d, CUDesc's "
                       "magic %u, CU's magic %u",
                       col, cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size, cu_desc_ptr->row_count,
                       cu_desc_ptr->magic, cu->m_magic);
        Assert(0);
    }

    // check the row number
    if (cu->m_offsetSize > 0) {
        // see also IMCU::FormValuesOffset()
        if ((cu->m_offsetSize / (int)sizeof(int32)) != (cu_desc_ptr->row_count + 1)) {
            HTAP_LOG_ERROR("HTAPFlushWorker: row_count mismatch in CU: relation info: name \"%s\", namespace id %u, "
                           "id %u, relfilenode %u/%u/%u",
                           RelationGetRelationName(m_table_ctx->m_rel), RelationGetNamespace(m_table_ctx->m_rel),
                           RelationGetRelid(m_table_ctx->m_rel), m_table_ctx->m_rel->rd_node.spcNode,
                           m_table_ctx->m_rel->rd_node.dbNode, m_table_ctx->m_rel->rd_node.relNode);
            HTAP_LOG_ERROR("HTAPFlushWorker: CU info: table column %d, id %u, offset %lu, size %d, magic %u, desc row "
                           "count %d, m_offset row count %d",
                           col, cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer, cu_desc_ptr->cu_size, cu_desc_ptr->magic,
                           cu_desc_ptr->row_count, ((cu->m_offsetSize / (int)sizeof(int32)) - 1));
            Assert(0);
        }
    }

    /* check cu size */
    if (cu->m_cuSize != (uint32)cu_desc_ptr->cu_size) {
        HTAP_LOG_ERROR("HTAPFlushWorker: cu size mismatch in CU: relation info: name \"%s\", namespace id %u, id %u, "
                       "relfilenode %u/%u/%u",
                       RelationGetRelationName(m_table_ctx->m_rel), RelationGetNamespace(m_table_ctx->m_rel),
                       RelationGetRelid(m_table_ctx->m_rel), m_table_ctx->m_rel->rd_node.spcNode,
                       m_table_ctx->m_rel->rd_node.dbNode, m_table_ctx->m_rel->rd_node.relNode);
        HTAP_LOG_ERROR(
            "CU info: table column %d, id %u, offset %lu, row count %d, magic %u. cu_desc size %d, cu_size %d", col,
            cu_desc_ptr->cu_id, cu_desc_ptr->cu_pointer, cu_desc_ptr->row_count, cu_desc_ptr->magic,
            (uint32)cu_desc_ptr->cu_size, cu->m_cuSize);
        Assert(0);
    }
}

}